package app.dwd;

import app.BaseSQLApp;
import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import utils.SQLUtil;

public class Dwd_03_DwdTradeWithSource extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdTradeWithSource().init(
                2303, 2,
                "Dwd_03_DwdTradeWithSource"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv, "Dwd_03_DwdTradeOrderDetail");
        //订单表
        Table orderDetail = tEnv.sqlQuery("select " +
                "data['id'] id,  " +
                "data['course_id'] course_id,  " +
                "data['order_id'] order_id,  " +
                "data['user_id'] user_id,  " +
                "data['origin_amount'] origin_amount,  " +
                "data['coupon_reduce'] coupon_reduce,  " +
                "data['final_amount'] final_amount,  " +
                "data['session_id'] session_id,  " +
                "data['create_time'] create_time,  " +
                "data['update_time'] update_time,  " +
                "ts  " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail", orderDetail);
        //日志来源表
        tEnv.executeSql("create table log_with_source(" +
                "`sc` string , " +
                "`uid` string, " +
                "`ts` bigint " +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRAFFIC_ACTION, "Dwd_03_DwdTradeOrderDetail"));
        Table logWithSource = tEnv.sqlQuery("select " +
                "sc sc,  " +
                "uid uid,  " +
                "ts  " +
                "from log_with_source ");
        tEnv.createTemporaryView("log_with_source", logWithSource);

        //join
        Table result = tEnv.sqlQuery("select " +
                "od.user_id, " +
                "od.order_id, " +
                "od.final_amount, " +
                "ls.sc, " +
                "od.ts, " +
                "current_row_timestamp() row_op_ts  " + /*得到每行计算的时间: 到后面 dws 去重的时候使用*/
                "from order_detail od " +
                "join log_with_source ls on od.user_id=ls.uid ");

        //kafka建表
        tEnv.executeSql("create table dwd_trade_with_source(  " +
                "user_id string, " +
                "order_id string, " +
                "final_amount string, " +
                "sc string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3)  " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_WITH_SOURCE));
        //写入kafka
        result.executeInsert("dwd_trade_with_source");

    }
}
